package kafka010;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SslConfigs;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;

public class KafkaProducerDemo {

    public static void main(String[] args) {
        send();
    }

    public static void send() {
//        String bootstrapAddress = "221.228.33.43:9092,221.228.33.43:9093,221.228.33.43:9094";
        String bootstrapAddress = "221.228.33.43:9096";
        String groupID = "test-tao";
        String topic = "test-topic-tao";
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "kafka010.DemoSerializer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "kafka010.DemoSerializer");

//构造 Producer 对象，注意，该对象是线程安全的，一般来说，一个进程内一个 Producer 对象即可；
//如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过 5 个
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        String value = "test msg from jsgx.telis" + new Date(); //消息的内容
        ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value);
        try {
            //发送消息，并获得一个 Future 对象
            Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
            //同步获得 Future 对象的结果
            RecordMetadata recordMetadata = metadataFuture.get();
            System.out.println(String.format("Produce ok: topic:%s partition:%d offset:%d value:%s",
                    recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset(), value));
        } catch (Exception e) {
            //要考虑重试
            //参考常见报错: https://help.aliyun.com/document_detail/68168.html?spm=a2c4g.11186623.6.567.2OMgCB
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }


}
